package com.education.realtime.app.dwd;

import com.education.realtime.app.BaseSQLApp;
import com.education.realtime.common.Constant;
import com.education.realtime.util.SqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdUserRegisterApp extends BaseSQLApp {
    public static void main(String[] args) {
        new DwdUserRegisterApp().init(2005,"DwdUserRegisterApp");
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {

        readOdsDb(tEnv,"DwdUserRegisterApp");

        Table table = tEnv.sqlQuery("select data['id'] user_id,\n" +
                "data['real_name'] user_name,\n" +
                "data['create_time'] create_time," +
                "ts\n" +
                "from ods_db\n" +
                "where `table` = 'user_info'\n" +
                "and type = 'insert' ");

        tEnv.executeSql("create table dwd_user_register(\n" +
                "user_id string,\n" +
                "user_name string,\n" +
                "create_time string,\n" +
                "ts string\n" +
                ")" + SqlUtil.getKafkaWriteDDL(Constant.TOPIC_DWD_USER_REGISTER));

        table.executeInsert("dwd_user_register");
    }
}
